package com.mall.manager.controller;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import com.mall.hadoop.format.AlphabetOutputFormat;
import com.mall.hadoop.init.HadoopInit;
import com.mall.hadoop.job.abilityrds.AbilityRdsMap;
import com.mall.hadoop.job.abilityrds.AbilityRdsReduce;
import com.mall.hadoop.job.abilitystatic.AbilityStaticMap;
import com.mall.hadoop.job.abilitystatic.AbilityStaticReduce;
import com.mall.hadoop.job.comjob.CommonJob;
import com.mall.hadoop.utils.HDFSUtil;
import com.mall.manager.service.AbilityStatistiService;
import com.mall.untils.DateUtils;
import com.mall.untils.PropertiesUtil;
import com.mongodb.DBObject;

/**
 * 八大能力统计
 * @author tgy
 *
 */
@RequestMapping("/manager")
@Controller
public class AbilityStatistiController {

	Logger logger = Logger.getLogger(AbilityStatistiController.class);
	
	@Resource
	AbilityStatistiService abilityStatistiService;
	
	
	/**
	 * 查询monogdb八大能力数据并上传到hdfs
	 * @return
	 */
	@RequestMapping("/findReviewInfo")
	@ResponseBody
	public String abilityUploadHdfs(){
		String fileName = "review_"+DateUtils.getDateFormat("yyyyMMdd");
		String res = "文件不存在";
		List<DBObject> review = abilityStatistiService.findMongodbReview("php_reviews");
		List<DBObject> school = abilityStatistiService.findMongodbSchool("php_school");
		List<DBObject> baby = abilityStatistiService.findMongodbBaby("php_baby");
		if(review.size()>0&&school.size()>0&&baby.size()>0){
			Map<String,DBObject> mapSchool = new HashMap<String,DBObject>();
			//遍历php_school表key:shoolId,value:DBObject
			for(DBObject dbo:school){
				mapSchool.put("school_"+dbo.get("id"),dbo);
			}
			Map<String,DBObject> mapBaby = new HashMap<String,DBObject>();
			//遍历php_baby表key:baby_id,value:school-DBObject
			for(DBObject dbo:baby){
				String schoolId = dbo.get("school_id").toString();
				mapBaby.put("baby_"+dbo.get("baby_id"), mapSchool.get("school_"+schoolId));
			}
			String path = PropertiesUtil.getProperties("filePath");
			File file1 = new File(path);
			//创建目录
			if(!file1.exists()){
				file1.mkdirs();
			}
			File file2 = new File(path+fileName+".txt");
			//创建文件
			if(!file2.exists()){
				try {
					file2.createNewFile();
				} catch (IOException e) {
					logger.info("创建文件（"+fileName+".txt）出现异常："+e.getMessage());
				}
			}
			//写入数据
			RandomAccessFile f = null;
			try {
				f = new RandomAccessFile(file2, "rw");
				for(DBObject re:review){
					//园所信息详情
					DBObject sch = mapBaby.get("baby_"+re.get("baby_id"));
					if(sch!=null){
						re.put("province",sch.get("province"));
						re.put("city", sch.get("city"));
						re.put("area", sch.get("area"));
						re.put("address", sch.get("address"));
						String str = re.toString()+"\r\n";
						f.write(str.getBytes("UTF-8"));
					}
				}
			} catch (Exception e) {
				logger.info("写入数据到文件（"+fileName+".txt）出现异常："+e.getMessage());
			}finally{
				try {
					f.close();
				} catch (IOException e) {
					logger.info("关闭写入流出现异常："+e.getMessage());
				}
			}
			//开始上传数据到hdfs
			try{
				Configuration conf = HadoopInit.getConfig();
				String filePath = path+fileName+".txt";
				String hdfsPath = PropertiesUtil.getProperties("dataHdfsInputPath");
				if(new File(filePath).exists()){				
					res = HDFSUtil.copyTxtToHdfs(conf, filePath, hdfsPath,fileName);
					//上传文件，删除文件
					if(res!=null){
						file2.delete();
						abilityStaticClear();
					}
				}else{
					logger.info("文件（"+fileName+".txt）不存在");
				}
			}catch(Exception e){
				logger.info("上传文件（"+fileName+".txt）到hdfs出现异常："+e.getMessage());
			}
		}
		return res;
	}
	
	/**
	 * 按地区统计八大能力
	 * @return
	 */
	@RequestMapping("/abilityClear")
	@ResponseBody
	public String abilityStaticClear(){
		String res = "文件不存在";
		try{
			String fileName = "review_"+DateUtils.getDateFormat("yyyyMMdd")+".txt";
			Configuration conf = HadoopInit.getConfig();
			String jobName = "abilityStatic";
			String input = PropertiesUtil.getProperties("dataHdfsInputPath")+fileName;
			String output = PropertiesUtil.getProperties("abilityHdfsOutputPath");
			res = CommonJob.commonJobStart(conf, jobName, input, output, AbilityStaticMap.class, AbilityStaticReduce.class, AlphabetOutputFormat.class);
			//分析成功后，插入到rds数据库
			if(res.equals("success")){
				res = insertRds(jobName,output+fileName);
			}
		}catch(Exception e){
			logger.info("按地区统计八大能力出现异常："+e.getMessage());
		}
	    return res;
	}
	
	/**
	 * 将数据插入数据库
	 * @param jobName
	 * @param fileName
	 * @return
	 */
	@RequestMapping("/insertAbility")
	@ResponseBody
	public String insertRds(String jobName,String fileName){
		String res = "文件不存在";
		try{
			Configuration conf = HadoopInit.getConfig();
			res = CommonJob.commonJobStart(conf, jobName, fileName, PropertiesUtil.getProperties("hdfsPath"), AbilityRdsMap.class, AbilityRdsReduce.class, AlphabetOutputFormat.class);
		}catch(Exception e){
			logger.info("将数据插入数据库出现异常："+e.getMessage());
		}
		return res;
	}
}
